/*
 * Anillo.cpp
 *
 *  Created on: 25/07/2014
 *      Author: nicolas
 */

#include "Anillo.h"
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <signal.h>
#include <errno.h>
#include <sstream>
#include <syslog.h>

using namespace std;

#define TIEMPO_ESPERA_RESPUESTA_INVITACION 5

typedef struct {
	unsigned int origen;
	unsigned int generacion;
} MSJ_ANILLO;

void timeOut_handler(int signo) {
	char mostrar[80];
	sprintf(mostrar, "Salida por timeout\n");
	write(fileno(stdout), mostrar, strlen(mostrar));
}

Anillo::Anillo(int puertoMulti, int puertoUni, int puertoTCP, char* ipMulticast,
		int time, int numeroBrokerLocal, int idBroker, ColaMensajes* cola,
		ColaMensajes* colaRespuesta) :
		shmem(numeroBrokerLocal) {
	sockMulti = socket(AF_INET, SOCK_DGRAM, 0);
	if (sockMulti < 0) {
		perror("Error en el socket UDP");
		exit(1);
	}
	sockUNI = socket(AF_INET, SOCK_DGRAM, 0);
	if (sockUNI < 0) {
		perror("Error en el socket UDP");
		exit(1);
	}
	this->puertoMulti = puertoMulti;
	this->puertoUNI = puertoUni;
	this->puertoTCP = puertoTCP;
	strcpy(this->ipMulticast, ipMulticast);
	configSockets();
	this->timeOutID = 0;
	this->time = time;
	this->cola = cola;
	this->colaRespuesta = colaRespuesta;
	this->idBroker = idBroker;
	this->numeroBrokerLocal = numeroBrokerLocal;
	idBrokerSiguiente = 0;
	esLider = false;
	esUnico = false;
	generacion = 0;
	tieneAnterior = false;
	tieneSiguienteEnvio = false;
	tieneSiguienteRecepcion = false;

	struct sigaction sa;
	memset(&sa, 0, sizeof(sa));
	sa.sa_handler = &timeOut_handler;
	sigaction(SIGUSR1, &sa, NULL);

	syslog(LOG_INFO, "Anillo con puertos multicast: %i, unicast: %i, TCP: %i",
			puertoMulti, puertoUni, puertoTCP);
}

void Anillo::enviarRespuesta(const char* msg) {
	MSJ_ANILLO mensaje;
	mensaje.generacion = generacion;
	mensaje.origen = idBroker;
	struct sockaddr_in addr;
	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_addr.s_addr = addrFrom.sin_addr.s_addr;
	addr.sin_port = htons(puertoUNI);
	if (sendto(sockUNI, &mensaje, sizeof(mensaje), 0, (sockaddr*) &addr,
			sizeof(addr)) < 0) {
		perror("Error en el send UDP, enviarRespuesta");
		exit(1);
	}
}

bool Anillo::obtenerConfirmacion() {
	lanzarTimeOut(TIEMPO_ESPERA_RESPUESTA_INVITACION);

	MSJ_ANILLO mensaje;
	unsigned int addLen;
	bool generacionActual = false;
	while (!generacionActual) {
		int mbytes = recvfrom(sockUNI, &mensaje, sizeof(mensaje), 0,
				(sockaddr*) &addrFrom, &addLen);
		if (mbytes < 0) {
			if (errno == EINTR) {
				return false;
			} else {
				perror("error en el recvfrom");
				exit(1);
			}
		}
		if (mensaje.generacion == generacion)
			generacionActual = true;
	}
	++timeOutID;
	shmem.set(timeOutID);
	return true;
}

void Anillo::enviarASiguiente(MENSAJE_BROKER& msg) {
	msg.destinatario = idBrokerSiguiente;
	siguienteEnvio.send(&msg, sizeof(MENSAJE_BROKER));
}

void Anillo::configSockets() {
	u_int yes = 1;
	if (setsockopt(sockMulti, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes))
			< 0) {
		perror("error en setsockopt(reuse)");
		exit(1);
	}
	struct ip_mreq mreq;
	memset(&mreq, 0, sizeof(mreq));
	mreq.imr_multiaddr.s_addr = inet_addr(ipMulticast);
	mreq.imr_interface.s_addr = htonl(INADDR_ANY );
	if (setsockopt(sockMulti, IPPROTO_IP, IP_ADD_MEMBERSHIP, &mreq,
			sizeof(mreq)) < 0) {
		perror("error en setsockopt(membership)");
		exit(1);
	}

	struct sockaddr_in addr;
	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_addr.s_addr = htonl(INADDR_ANY );
	addr.sin_port = htons(puertoMulti);
	if (bind(sockMulti, (const struct sockaddr *) &addr, sizeof(addr)) < 0) {
		perror("error en el bind UDP, multicast");
		exit(1);
	}
	///----------------

	if (setsockopt(sockUNI, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) < 0) {
		perror("error en setsockopt(reuse)");
		exit(1);
	}

	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_addr.s_addr = htonl(INADDR_ANY ); //addrFrom.sin_addr.s_addr;
	addr.sin_port = htons(puertoUNI);
	if (bind(sockUNI, (const struct sockaddr *) &addr, sizeof(addr)) < 0) {
		perror("error en el bind UDP, obtener confirmacion");
		exit(1);
	}

}

void Anillo::conectarTCP(const char* ip) {

	handshakeTCP(siguienteEnvio, idBroker, ip, puertoTCP, true);

	handshakeTCP(siguienteRecepcion, idBroker, ip, puertoTCP, false);

	tieneSiguienteEnvio = true;
	tieneSiguienteRecepcion = true;

	syslog(LOG_INFO, "Coneccion establecida");
}

void Anillo::interconectar() {
	if (esUnico)
		return;

	syslog(LOG_INFO, "Inicio interconeccion");

	MENSAJE_BROKER msg;
	struct sockaddr addr;
	int portAux;
	char ip[16];

	msg.id = idBroker;
	msg.tipo = TIPO_INTERCONECCION_ANILLO;

	if (esLider) {
		enviarASiguiente(msg);
	} else {
		cola->recibir(&msg, TAM_MENSAJE_BROKER, idBroker);

		syslog(LOG_INFO, "Conectando con %i", msg.id);

		rpc.datosDeBroker(msg.id, addr, portAux);
		copiarIP(addr, ip);
		Socket sockAux;
		handshakeTCP(sockAux, idBroker, ip, portAux, false);
		lanzarConeccionSimpleTCP(sockAux.getFd());
		sockAux.closeSock();

		int aux = msg.id;

		msg.id = idBroker;
		enviarASiguiente(msg);

		msg.id = aux;
		enviarASiguiente(msg);
	}

	bool salir = false;
	while (!salir) {
		cola->recibir(&msg, TAM_MENSAJE_BROKER, idBroker);
		if (msg.id == idBroker) {
			salir = true;
		} else {
			syslog(LOG_INFO, "Conectando con %i", msg.id);

			rpc.datosDeBroker(msg.id, addr, portAux);
			copiarIP(addr, ip);
			Socket sockAux;
			handshakeTCP(sockAux, idBroker, ip, portAux, false);
			lanzarConeccionSimpleTCP(sockAux.getFd());
			sockAux.closeSock();

			enviarASiguiente(msg);
		}
	}

	syslog(LOG_INFO, "Fin interconeccion");
}

void Anillo::copiarIP(struct sockaddr& addr, char* ip) {
	struct sockaddr_in addrAux;
	memcpy(&addrAux, &addr, sizeof(addr));
	strcpy(ip, inet_ntoa(addrAux.sin_addr));
}

void Anillo::lanzarConeccionSimpleTCP(int socketFd) {
	int pid = fork();
	stringstream fdStr, idStr, receptStr;
	fdStr << socketFd;
	idStr << idBroker;
	receptStr << 1;
	if (pid == 0) {
		execlp("./brokerComunicator", "brokerComunicator", fdStr.str().c_str(),
				idStr.str().c_str(), receptStr.str().c_str(), NULL);
		perror("Error en el exec brokerComunicator");
		exit(1);
	}
}

bool Anillo::recuperarToken(unsigned int version, unsigned int generacion) {
	char mostrar[80];
	sprintf(mostrar, "Recuperando Token\n");
	write(fileno(stdout), mostrar, strlen(mostrar));
	syslog(LOG_INFO, "Recuperando Token");

	if (esUnico)
		return true;

	bool tieneAlMasReciente = false;
	MENSAJE_BROKER msg;

	if (esLider) {
		msg.version = version;
		msg.generacion = generacion;
		msg.origen = idBroker;
		enviarASiguiente(msg);
	} else {
		cola->recibir(&msg, TAM_MENSAJE_BROKER, idBroker);
		if (tokenPropioMasReciente(version, generacion, msg))
			tieneAlMasReciente = true;
		else
			tieneAlMasReciente = false;

		unsigned int versAux, genAux;

		versAux = msg.version;
		genAux = msg.generacion;

		msg.version = version;
		msg.generacion = generacion;

		enviarASiguiente(msg);

		msg.version = versAux;
		msg.generacion = genAux;

		enviarASiguiente(msg);

	}

	bool salir = false;

	while (!salir) {
		cola->recibir(&msg, TAM_MENSAJE_BROKER, idBroker);
		if (msg.version == version && msg.generacion == generacion) {
			salir = true;
		} else {
			if (tokenPropioMasReciente(version, generacion, msg))
				tieneAlMasReciente = true;
			else
				tieneAlMasReciente = false;
			enviarASiguiente(msg);
		}
	}

	return tieneAlMasReciente;
}

bool Anillo::tokenPropioMasReciente(unsigned int version,
		unsigned int generacion, MENSAJE_BROKER& msg) {
	if (msg.generacion == generacion) {
		if (msg.version < version) {
			return true;
		}
	} else {
		if (msg.version > version) {
			return true;
		}
	}

	return false;

}

Anillo::~Anillo() {
	MENSAJE_BROKER msg;
	msg.tipo = TIPO_FINALIZAR;

	if (tieneSiguienteEnvio)
		siguienteEnvio.send(&msg, sizeof(MENSAJE_BROKER));
	if (tieneSiguienteRecepcion)
		siguienteRecepcion.send(&msg, sizeof(MENSAJE_BROKER));
	if (tieneAnterior)
		anterior.send(&msg, sizeof(MENSAJE_BROKER));

	siguienteEnvio.closeSock();
	siguienteRecepcion.closeSock();
	anterior.closeSock();

}

void Anillo::enviarASiguiente(MENSAJE_TOKEN& msg) {
	msg.destinatario = idBrokerSiguiente;
	siguienteEnvio.send(&msg, sizeof(MENSAJE_TOKEN));
}

void Anillo::handshakeTCP(Socket& socket, int id, const char* ip, int puerto,
		bool brokerEsReceptor) {
	MENSAJE_BROKER msg;

	socket.connectTo(ip, puerto);
	msg.esReceptor = brokerEsReceptor;
	msg.id = id;
	socket.send(&msg, sizeof(MENSAJE_BROKER));

}

void Anillo::generar(int generacion) {

	char mostrar[80];
	sprintf(mostrar, "Generando anillo\n");
	write(fileno(stdout), mostrar, strlen(mostrar));

	MENSAJE_BROKER msg;

	bool conectado = false;
	bool fueInvitado = true;
	esLider = false;
	char liderIP[16];
	int liderPuerto;

	while (!conectado && fueInvitado) {
		sprintf(mostrar, "Esperando invitacion\n");
		write(fileno(stdout), mostrar, strlen(mostrar));
		syslog(LOG_INFO, "Esperando invitacion");
		fueInvitado = esperarMensaje(sockMulti, time);
		if (fueInvitado) {
			sprintf(mostrar, "Invitacion recibida\n");
			write(fileno(stdout), mostrar, strlen(mostrar));
			syslog(LOG_INFO, "Invitacion recibida");
			enviarRespuesta("Me uno");
			bool confirmado = obtenerConfirmacion();
			if (confirmado) {
				sprintf(mostrar, "Confirmacion recibida\n");
				write(fileno(stdout), mostrar, strlen(mostrar));
				syslog(LOG_INFO, "Confirmacion recibida");
				char ip[16];
				strcpy(ip, inet_ntoa(addrFrom.sin_addr));
				conectarTCP(ip);
				msg.destinatario = DESTINATARIO_ANILLO;
				msg.origen = idBroker;
				siguienteEnvio.send(&msg, sizeof(MENSAJE_BROKER));
				siguienteRecepcion.recv(&msg, sizeof(MENSAJE_BROKER));
				msg.ip[15] = '\0';

				idBrokerSiguiente = msg.id;

				if (msg.primerConeccion) {
					strcpy(liderIP, ip);
					liderPuerto = puertoTCP;
				} else {
					strcpy(liderIP, msg.ip);
					liderPuerto = msg.puerto;
				}
				conectado = true;
			} else {
				sprintf(mostrar, "Sin confirmacion\n");
				write(fileno(stdout), mostrar, strlen(mostrar));
				syslog(LOG_INFO, "Sin confirmacion");
			}
		}
	}

	if (!fueInvitado) {
		esLider = true;
	}

	bool aceptadaInvitacion = false;
	int reintentos = 3;

	while (!aceptadaInvitacion && reintentos > 0) {
		reintentos--;
		sprintf(mostrar, "Lanzando invitacion %i\n", reintentos);
		write(fileno(stdout), mostrar, strlen(mostrar));
		lanzarInvitacion();
		aceptadaInvitacion = esperarMensaje(sockUNI,
				TIEMPO_ESPERA_RESPUESTA_INVITACION);
	}

	if (aceptadaInvitacion) {
		sprintf(mostrar, "invitacion aceptada\n");
		write(fileno(stdout), mostrar, strlen(mostrar));
		enviarRespuesta("Aceptado");
		cola->recibir(&msg, TAM_MENSAJE_BROKER, DESTINATARIO_ANILLO);
		msg.destinatario = msg.origen;
		msg.origen = idBroker;
		msg.id = idBroker;
		if (esLider)
			msg.primerConeccion = true;
		else {
			strcpy(msg.ip, liderIP);
			msg.primerConeccion = false;
		}
		msg.puerto = puertoTCP;
		colaRespuesta->enviar(&msg, TAM_MENSAJE_BROKER);
		if (esLider) {
			cola->recibir(&msg, TAM_MENSAJE_BROKER, DESTINATARIO_ANILLO);
			idBrokerSiguiente = msg.id;
			struct sockaddr addr;
			int portAux;
			rpc.datosDeBroker(idBrokerSiguiente, addr, portAux);

			char ip[16];
			copiarIP(addr, ip);

			handshakeTCP(siguienteEnvio, idBroker, ip, puertoTCP, true);
			tieneSiguienteEnvio = true;

			syslog(LOG_INFO, "Anillo cerrado por ip: %s", ip);

		}
	} else {
		//Cerrar el anillo
		sprintf(mostrar, "Cerrando anillo\n");
		write(fileno(stdout), mostrar, strlen(mostrar));
		if (esLider) {
			//UNICO BROKER EN ANILLO
			esUnico = true;

			strcpy(liderIP, "localhost");
			liderPuerto = puertoTCP;

			handshakeTCP(siguienteEnvio, idBroker, liderIP, liderPuerto, true);
			tieneSiguienteEnvio = true;

			idBrokerSiguiente = idBroker;

		} else {
			syslog(LOG_INFO, "Cerrando anillo a ip: %s, puerto: %i", liderIP,
					liderPuerto);

			handshakeTCP(anterior, idBroker, liderIP, liderPuerto, true);

			msg.destinatario = DESTINATARIO_ANILLO;
			msg.id = idBroker;
			anterior.send(&msg, sizeof(MENSAJE_BROKER));
			tieneAnterior = true;

		}
	}
}

bool Anillo::esperarMensaje(int sock, int tiempo) {
	lanzarTimeOut(tiempo);
	MSJ_ANILLO mensaje;
	unsigned int addLen = sizeof(addrFrom);
	bool generacionActual = false;
	while (!generacionActual) {
		int mbytes = recvfrom(sock, &mensaje, sizeof(mensaje), 0,
				(sockaddr*) &addrFrom, &addLen);
		if (mbytes < 0) {
			if (errno == EINTR) {
				return false;
			} else {
				perror("error en el recvfrom, esperar mensaje");
				exit(1);
			}
		}
		if (mensaje.generacion == generacion)
			generacionActual = true;
	}
	++timeOutID;
	shmem.set(timeOutID);
	return true;
}

void Anillo::generarToken() {

	if (!esLider)
		return;

	char mostrar[80];

	sprintf(mostrar, "Generando Token\n");
	write(fileno(stdout), mostrar, strlen(mostrar));
	syslog(LOG_INFO, "Generando Token");

	MENSAJE_BROKER msg;
	MENSAJE_TOKEN tokenMsg;

	TOKEN tok;
	memset(&(tok.parque), 0, sizeof(PARQUE));
	memset(&(tok.salasEntrada), 0, sizeof(SALA) * CANT_ENTRADAS);
	memset(&(tok.salasZoo), 0, sizeof(SALA) * CANT_ENTRADAS);
	memset(&(tok.semBus), 0, sizeof(bool) * CANT_ENTRADAS);
	memset(&(tok.semSalasEntrada), 1, sizeof(bool) * CANT_ENTRADAS);
	memset(&(tok.semSalasZoo), 1, sizeof(bool) * CANT_ENTRADAS);
	memset(&(tok.semSalasLlenoEntrada), 0, sizeof(bool) * CANT_ENTRADAS);
	memset(&(tok.semSalasLlenoZoo), 0, sizeof(bool) * CANT_ENTRADAS);
	memset(&(tok.semParque), 1, sizeof(bool));
	tok.parque.busesDeRepuestos = CANT_BUSES_REPUESTO;
	tok.generacion = 0;
	tok.version = 0;

	for (int i = 0; i < CANT_ENTRADAS; ++i) {
		tok.salasEntrada[i].estadoBUS = Tipos::BUS_ACTIVO;
		tok.salasZoo[i].estadoBUS = Tipos::BUS_ACTIVO;
	}

	msg.origen = idBroker;
	msg.tipo = TIPO_TOKEN;
	enviarASiguiente(msg);

	tokenMsg.origen = idBroker;

	memcpy(&(tokenMsg.token), &tok, sizeof(TOKEN));
	enviarASiguiente(tokenMsg);
}

void Anillo::lanzarTimeOut(int tiempo) {
	timeOutID++;
	shmem.set(timeOutID);
	stringstream id, time, broker;
	id << timeOutID;
	time << tiempo;
	broker << this->numeroBrokerLocal;
	int pid = fork();
	if (pid == 0) {
		execlp("./timeOut", "timeOut", id.str().c_str(), time.str().c_str(),
				broker.str().c_str(), NULL);
		perror("Error en el exec timeout");
		exit(1);
	}

}

void Anillo::lanzarTimeOutToken(int tiempo) {
	timeOutID++;
	shmem.set(timeOutID);
	stringstream id, time, brokerLocal, brokerGlobal;
	id << timeOutID;
	time << tiempo;
	brokerLocal << this->numeroBrokerLocal;
	brokerGlobal << this->idBroker;
	int pid = fork();
	if (pid == 0) {
		execlp("./timeOutToken", "timeOutToken", id.str().c_str(),
				time.str().c_str(), brokerLocal.str().c_str(),
				brokerGlobal.str().c_str(), NULL);
		perror("Error en el exec timeout");
		exit(1);
	}
}

void Anillo::lanzarInvitacion() {
	MSJ_ANILLO mensaje;
	mensaje.generacion = generacion;
	mensaje.origen = idBroker;
	struct sockaddr_in addr;
	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_addr.s_addr = inet_addr(ipMulticast);
	addr.sin_port = htons(puertoMulti);
	if (sendto(sockMulti, &mensaje, sizeof(mensaje), 0, (sockaddr*) &addr,
			sizeof(addr)) < 0) {
		perror("Error en el send UDP, lanzarInvitacion");
		exit(1);
	}
}

